package main

import (
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/IBM/sarama"
)

const (
	brokerAddress = "localhost:9092"
	topic         = "test_topic"
	group         = "my-consumer-group"
)

func main() {
	// 创建 Kafka 消费者配置
	config := sarama.NewConfig()
	config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
	config.Consumer.Offsets.Initial = sarama.OffsetOldest

	// 创建 Kafka 消费者
	consumer, err := sarama.NewConsumer([]string{brokerAddress}, config)
	if err != nil {
		log.Fatalln("Failed to start Kafka consumer:", err)
	}
	defer consumer.Close()

	// 创建 Kafka 分区消费者
	partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetNewest)
	if err != nil {
		log.Fatalln("Failed to start partition consumer:", err)
	}
	defer partitionConsumer.Close()

	// 创建信号通道，以便在收到终止信号时优雅地退出
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)

	// 从分区消费者获取消息和消费错误
	consumeErrors := partitionConsumer.Errors()

	for {
		select {
		case msg := <-partitionConsumer.Messages():
			// 收到消息
			fmt.Printf("Received message: %s\n", string(msg.Value))

			// 在这里添加消费消息的处理逻辑，比如存储到数据库或执行业务逻辑
			// 注意：在处理消息时，需要保证处理的幂等性，以防止重复处理相同的消息

			time.Sleep(3 * time.Second) // 假设处理消息需要 3 秒

			fmt.Println("Message processed successfully")

			// 标记消息为已处理，以便在下次启动时从正确的位置继续消费
			partitionConsumer.MarkOffset(msg, "")

		case err := <-consumeErrors:
			// 消费消息时出现错误
			fmt.Fprintf(os.Stderr, "Failed to consume message: %s\n", err)

			// 在这里添加失败后的处理逻辑，比如日志记录、指数退避等

			time.Sleep(1 * time.Second) // 1 秒后重试
		case <-signals:
			// 收到终止信号，优雅地退出
			fmt.Println("Received termination signal. Exiting...")
			return
		}
	}
}
